package com.shujia.wyh.jinjie;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.List;

/*
    利用已经构建好的索引表，模拟查询数据的过滤

    students_index  rowkey:name  列名:学号

    students  name,age,gender,clazz

    需求：查询某个学生在哪个班级
 */
public class HbaseIndexDemo2 {
    private static Connection conn;
    private static Admin admin;

    public static void main(String[] args) throws Exception {
        //创建表
        //获取配置文件对象
        Configuration conf = new Configuration();
        //2、设置以下zookeeper的集群信息
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();

        System.out.println("获取连接成功！！" + conn);
        System.out.println("HMaster的对象获取成功！！" + admin);
        System.out.println("------------------------------------------------------------------");

//
//        getData1(); // 456 毫秒
        getData2();   // 502 毫秒


    }

    public static void getData2() throws Exception {
        long start = System.currentTimeMillis();
        //方小蕾
        TableName students = TableName.valueOf("students");

        Table table2 = conn.getTable(students);

        Scan scan = new Scan();

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes("info"),
                Bytes.toBytes("name"),
                CompareOperator.EQUAL,
                new BinaryComparator(Bytes.toBytes("方小蕾")));

        scan.setFilter(singleColumnValueFilter);
        ResultScanner scanner = table2.getScanner(scan);

        Result rs = null;
        while ((rs = scanner.next()) != null) {
            List<Cell> cells = rs.listCells();
            for (Cell cell : cells) {
                String s = Bytes.toString(CellUtil.cloneQualifier(cell));
                if ("clazz".equals(s)) {
                    String clazz = Bytes.toString(CellUtil.cloneValue(cell));
                    String id = Bytes.toString(CellUtil.cloneRow(cell));
                    System.out.println("姓名为：方小蕾 学号为：" + id + ", 班级为：" + clazz);
                    break;
                }
            }
        }

//        String id = Bytes.toString(scanner.next().getRow());
//
//        Get get = new Get(Bytes.toBytes(id));
//
//        Result result = table2.get(get);
//        String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));


        long end = System.currentTimeMillis();
        System.out.println("普通查询方式，共耗时：" + (end - start) + " 毫秒");
    }

    public static void getData1() throws Exception {
        long start = System.currentTimeMillis();
        //方小蕾
        TableName studentsIndex = TableName.valueOf("students_index");
        TableName students = TableName.valueOf("students");

        Table table = conn.getTable(studentsIndex);

        Table table2 = conn.getTable(students);
        Get get = new Get(Bytes.toBytes("方小蕾"));

        Result result = table.get(get);

        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            String id = Bytes.toString(CellUtil.cloneQualifier(cell));
            //拿着学号去查询原表
            Get get1 = new Get(Bytes.toBytes(id));
            Result result1 = table2.get(get1);

            String clazz = Bytes.toString(result1.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));

            System.out.println("姓名为：方小蕾 学号为：" + id + ", 班级为：" + clazz);
        }
        long end = System.currentTimeMillis();
        System.out.println("查索引数据方式，共耗时：" + (end - start) + " 毫秒");
    }
}
